package ai.people.netmon.userconfig.service.impl;

import ai.people.core.standard.gather.userconfig.UserConfigSyncService;
import ai.people.netmon.framework.businessenum.AnalysisSourceEnum;
import ai.people.netmon.framework.businessenum.gather.GatherStatusEnum;
import ai.people.netmon.framework.constant.SourceCollectionConstant;
import ai.people.netmon.framework.domain.gather.config.GatherSyncExtended;
import ai.people.netmon.framework.domain.userconfig.entity.UserCollectConfigAccount;
import ai.people.netmon.framework.domain.userconfig.entity.UserCollectConfigAccountPartner;
import ai.people.netmon.framework.domain.userconfig.request.AccountQueryRequest;
import ai.people.netmon.framework.exception.enums.CommonEnum;
import ai.people.netmon.framework.exception.enums.SourceExceptionEnum;
import ai.people.netmon.framework.exception.enums.URLExceptionEnum;
import ai.people.netmon.framework.exception.type.BusinessException;
import ai.people.netmon.framework.model.response.PageResponse;
import ai.people.netmon.framework.utils.AssertUtils;
import ai.people.netmon.userconfig.mapper.UserCollectConfigAccountMapper;
import ai.people.netmon.userconfig.service.DataSyncService;
import ai.people.netmon.userconfig.service.UserCollectConfigAccountPartnerService;
import ai.people.netmon.userconfig.service.UserCollectConfigAccountService;
import ai.people.netmon.userconfig.service.handle.HandleUrlService;
import ai.people.netmon.userconfig.service.handle.HandleUrlUtil;
import ai.people.netmon.utils.UrlUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.stream.Collectors;

import static java.util.stream.Collectors.*;

/**
 * <p>
 * （监测系统）用户配置采集账号表 服务实现类
 * </p>
 *
 * @author zhenglin
 * @since 2022-02-28
 */
@Service
@Slf4j
public class UserCollectConfigAccountServiceImpl extends ServiceImpl<UserCollectConfigAccountMapper, UserCollectConfigAccount> implements UserCollectConfigAccountService, UserConfigSyncService<UserCollectConfigAccount> {

    @Autowired
    UserCollectConfigAccountMapper mapper;
    @Autowired
    DataSyncService dataSyncService;
    @Value("${data-sync.account-url}")
    private String accountSyncUrl;
    @Autowired
    KafkaTemplate<String, String> kafkaTemplate;
    @Autowired
    UserCollectConfigAccountPartnerService userCollectConfigAccountPartnerService;

    /**
     * 分页查询
     *
     * @param request 请求
     * @return {@link PageResponse}<{@link UserCollectConfigAccount}>
     */
    @Override
    public PageResponse<UserCollectConfigAccount> pageQuery(AccountQueryRequest request) {
        Page<UserCollectConfigAccount> page = new Page<>(request.getOriginPageNo(), request.getPageSize());
        LambdaQueryWrapper<UserCollectConfigAccount> queryWrapper = request.createQueryWrapper();
        Page<UserCollectConfigAccount> result = page(page, queryWrapper);
        return PageResponse.builder(result);
    }

    /**
     * 通过id获取帐户
     *
     * @param id id
     * @return {@link UserCollectConfigAccount}
     */
    @Override
    public UserCollectConfigAccount getAccountById(Long id) {
        return getById(id);
    }

    /**
     * 保存账户
     *
     * @param account 账户
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void saveAccount(UserCollectConfigAccount account) {

        preHandle(account);
        boolean save = save(account);
        AssertUtils.isTrue(save, CommonEnum.FAIL);
        UserCollectConfigAccountPartner userCollectConfigAccountPartner = new UserCollectConfigAccountPartner(account.getAccount(), account.getPartnerId());
        boolean save1 = userCollectConfigAccountPartnerService.save(userCollectConfigAccountPartner);
        AssertUtils.isTrue(save1, CommonEnum.FAIL);
//        syncAccount(new UserConfigAccountData(account),account.getInfoType(), MonitorUserConfigTopic.MONITOR_USER_CONFIG_TOPIC_ACCOUNTS.name(),kafkaTemplate,new GatherSyncExtended(),()->updateById(account));

        //推送
        GatherSyncExtended gatherSyncExtended = getGatherSyncExtended(account);
        syncAccounts(Collections.singletonList(account), account.getInfoType(), accountSyncUrl, gatherSyncExtended);

        dataSyncService.sendToDiffSource(Collections.singletonList(account));
    }

    /**
     * 前处理
     *
     * 解析url，判断链接是否是系统信源，并获取链接中的账号ID
     *
     * @param account 账户
     */
    private void preHandle(UserCollectConfigAccount account) {
        String host = UrlUtil.getHost(account.getAccountLink());
        if (StringUtils.isBlank(host)){
            throw new BusinessException(URLExceptionEnum.CANNOT_PARSE);
        }
        if (!AnalysisSourceEnum.check(host)){
            throw new BusinessException(SourceExceptionEnum.SOURCE_NOT_EXISTS);
        }

        account.setInfoType(host);
        HandleUrlService handleUrlService = HandleUrlUtil.getService(host);
        String accountId = handleUrlService.handleUrl(account);

        if (StringUtils.isNotBlank(accountId)) {
            account.setAccount(accountId);
        }
        if (StringUtils.isBlank(account.getCrawlInterval())) {
            account.setCrawlInterval(SourceCollectionConstant.COLLECTION_FREQUENCY_STR);
        }
        account.setCollectionStatus(GatherStatusEnum.TO_BE_COLLECTED.getCode());
        account.setPartnerId(account.getUserId());
    }

    private GatherSyncExtended getGatherSyncExtended(UserCollectConfigAccount account) {
        return getGatherSyncExtended(Collections.singletonList(account));
    }

    private GatherSyncExtended getGatherSyncExtended(List<UserCollectConfigAccount> accounts) {

        Set<String> accountNameSet = accounts.stream().map(UserCollectConfigAccount::getAccount).collect(Collectors.toSet());

        //查询关键词的partner id
        LambdaQueryWrapper<UserCollectConfigAccountPartner> queryWrapper = Wrappers.lambdaQuery();
        queryWrapper.select(UserCollectConfigAccountPartner::getAccountName, UserCollectConfigAccountPartner::getPartnerId);
        queryWrapper.in(UserCollectConfigAccountPartner::getAccountName, accountNameSet);
        List<UserCollectConfigAccountPartner> list = userCollectConfigAccountPartnerService.list(queryWrapper);

        Map<String, String> gpByAccountName = list.stream().collect(groupingBy(UserCollectConfigAccountPartner::getAccountName,
                collectingAndThen(toList(), x -> x.stream().distinct().map(UserCollectConfigAccountPartner::getPartnerId).map(y -> "\"" + y + "\"").collect(joining(",")))));
        //构造
        GatherSyncExtended gatherSyncExtended = new GatherSyncExtended();
        for (Map.Entry<String, String> entry : gpByAccountName.entrySet()) {
            entry.setValue("[" + entry.getValue() + "]");
        }
        gatherSyncExtended.setPidMapping(gpByAccountName);
        return gatherSyncExtended;
    }

    /**
     * 批量保存账户
     *
     * @param list 列表
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void saveAccountBatch(List<UserCollectConfigAccount> list) {
        if (CollectionUtils.isEmpty(list)) {
            return;
        }
        list.forEach(this::preHandle);
        //保存账号
        boolean b = saveBatch(list);
        AssertUtils.isTrue(b, CommonEnum.FAIL);
        //保存账号和partner的关系
        List<UserCollectConfigAccountPartner> partners = new ArrayList<>();
        list.forEach(x -> partners.add(new UserCollectConfigAccountPartner(x.getAccount(), x.getPartnerId())));
        boolean b1 = userCollectConfigAccountPartnerService.saveBatch(partners);
        AssertUtils.isTrue(b1, CommonEnum.FAIL);

        GatherSyncExtended gatherSyncExtended = getGatherSyncExtended(list);

        Map<String, List<UserCollectConfigAccount>> collect = list.stream().collect(groupingBy(UserCollectConfigAccount::getInfoType));
        collect.forEach((k, v) -> {
//            List<UserConfigAccountData> accountDataList = new ArrayList<>();
//            v.forEach(x -> accountDataList.add(new UserConfigAccountData(x)));
//            syncAccount(accountDataList, k, MonitorUserConfigTopic.MONITOR_USER_CONFIG_TOPIC_ACCOUNTS.name(), kafkaTemplate, new GatherSyncExtended(), () -> updateBatchById(v));
            syncAccounts(v, k, accountSyncUrl, gatherSyncExtended);
            dataSyncService.sendToDiffSource(v);
        });
    }

    /**
     * 更新账号
     *
     * @param account 账号
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void updateAccount(UserCollectConfigAccount account) {
        AssertUtils.isTrue(updateById(account), CommonEnum.FAIL);
    }

    /**
     * 通过id删除账号
     *
     * @param id id
     * @return boolean
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void delAccountById(Long id) {
        AssertUtils.isTrue(removeById(id), CommonEnum.FAIL);
    }


    @Override
    public void updateCollectionStatus(List<UserCollectConfigAccount> account) {
        account.forEach(x -> x.setCollectionStatus(GatherStatusEnum.TO_BE_COLLECTED.getCode()));
        updateBatchById(account);
    }
}
